package com.ccb.channel.pos;

import org.apache.log4j.Logger;

import java.io.IOException;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import com.alibaba.fastjson.JSON;
import com.ccb.util.Record;
import com.ccb.util.RecordType;
import com.ccb.util.StandardRecord;

public class PosReducer extends Reducer<Text, Text, Text, NullWritable> {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(PosReducer.class);

	private MultipleOutputs<Text, NullWritable> mos = null;

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("setup(Context) - start");
		}

		mos = new MultipleOutputs<Text, NullWritable>(context);

		if (logger.isDebugEnabled()) {
			logger.debug("setup(Context) - end");
		}
	}

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("cleanup(Context) - start");
		}

		mos.close();

		if (logger.isDebugEnabled()) {
			logger.debug("cleanup(Context) - end");
		}
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("reduce(Text, Iterable<Text>, Context) - start");
		}

		StandardRecord standardRecord = new StandardRecord();
		for (Text val : values) {
			Record rd = JSON.parseObject(val.toString(), Record.class);
			standardRecord.setUid(rd.getUid());
			standardRecord.setServiceId(rd.getServiceId());
			if (rd.getRecordType().equals(RecordType.REQUEST)) {
				standardRecord.setRequestData(rd.getData());
				standardRecord.setRequest(rd.getOriginalData());
				standardRecord.setRecordType(RecordType.REQUEST);
				standardRecord.setTime(rd.getTime());
			}
			if (rd.getRecordType().equals(RecordType.RESPONSE)) {
				standardRecord.setResponseData(rd.getData());
				standardRecord.setResponse(rd.getOriginalData());
				standardRecord.setRecordType(RecordType.RESPONSE);
				standardRecord.setResponseCode(rd.getResponseCode());
				
			}
		}
		if (standardRecord.getRequest() != null && standardRecord.getResponse() != null)
			standardRecord.setRecordType(RecordType.BOTH);
		mos.write( new Text(JSON.toJSONString(standardRecord)), NullWritable.get(),  standardRecord.getServiceId());
		//context.write(new Text(JSON.toJSONString(standardRecord)), null);

		if (logger.isDebugEnabled()) {
			logger.debug("reduce(Text, Iterable<Text>, Context) - end");
		}
	}

}
